Spark Tuning - Advanced Partitioning: Theoretical Quiz

This assessment details partitioning strategies, salting patterns for data skew, and bucketing mechanics.

Scenario 1: Salting Technique to Optimize Data Skew

The Scenario

A large retail firm runs a daily transactional join: sales_df.join(items_df, "item_id") A small group of hot items (e.g. "promo_deal_99") accounts for 80% of all sales sales data volume. Executors processing these hot keys crash with JVM OOM exceptions or run 30x slower than other nodes.

The Questions

1. Explain how Salting acts as an architectural workaround to mitigate data skew joins.
2. Provide a PySpark code implementation of salting for a highly skewed join.

Detailed Solution & Architectural Analysis

1. Salting Mechanics

Data skew join bottleneck occurs because all sales records containing item_id = "promo_deal_99" shuffle to a single reducer executor.

• Salt Prefixing: We append a random integer (the "salt", e.g., 0 to 9) to the join key of the skewed table (sales_df): "promo_deal_99" "promo_deal_99_3"

• Exploding the Lookup Table: To ensure joins still match, we explode the lookup table (items_df) so that each original item key is replicated 10 times, appended with suffixes 0 to 9.

• Uniform Partition Distribution: The sales records are now distributed across 10 distinct executors in parallel, completely resolving memory hotspots and boosting performance.

2. Salting PySpark Implementation

import pyspark.sql.functions as F

# 1. Add random salt (0 to 4) to skewed Sales DataFrame
salted_sales = sales_df.withColumn("salt", F.concat(F.col("item_id"), F.lit("_"), F.randint(0, 4)))

# 3. Replicate Lookup Items DataFrame 5 times (0 to 4) to match salts
replicated_items = items_df.withColumn("salt_array", F.array([F.lit(i) for i in range(5)])) \
                            .withColumn("salt_val", F.explode("salt_array")) \
                            .withColumn("salted_join_key", F.concat(F.col("item_id"), F.lit("_"), F.col("salt_val")))

# 4. Execute the join on salted keys
result_df = salted_sales.join(replicated_items, salted_sales.salt == replicated_items.salted_join_key)

Scenario 2: Bucketing vs. Partitioning

The Scenario

An architect designs an analytics data lake. The queries frequently filter by country and join tables on user_id.

The Questions

1. Differentiate between file Partitioning and Bucketing in terms of folder layout, file constraints, and join optimizations.
2. Under what exact cardinality circumstances should bucketing be preferred?

Detailed Solution & Architectural Analysis

1. Partitioning vs. Bucketing

2. Optimal Bucketing Selection

Bucketing should be preferred when:

1. High Cardinality Columns: The target column has thousands or millions of unique values (e.g. user_id, device_id). Creating folder-based partitioning on these would generate "many tiny files" metadata crashes in HDFS/S3.
2. Frequent Large Joins: Large tables are joined constantly on the bucket key. Pre-shuffling and pre-sorting them into a fixed number of buckets at write time allows Spark to join them downstream with zero network shuffles.

Scenario 3: Dynamic Partition Pruning (DPP)

The Scenario

A query joins a large fact table partitioned by date_key with a tiny, filtered dimension table. The execution plan reports: INFO: DynamicPartitionPruning: Skipping 340 partitions from Fact table

The Questions

1. Explain the architectural mechanics of Dynamic Partition Pruning (DPP).
2. What are the key plan requirements to trigger DPP during joins?

Detailed Solution & Architectural Analysis

1. DPP Mechanics

In traditional joins, Spark scans the entire partitioned fact table before applying the join filter.

• The Optimization: DPP allows Spark to filter partitions at compile/scan time based on filter results from the dimension table.
• How it works: Spark runs a quick subquery on the filtered dimension table to extract the active date_key values. It then passes these keys directly to the Fact table scan, pruning (skipping) irrelevant directory partitions before loading data, saving massive disk read operations.

2. DPP Requirements

To trigger DPP:

1. The fact table must be physically partitioned on the join key (date_key).
2. The join must be an Equi-Join (matching exact values using =).
3. The join strategy must be a Broadcast Join (so the filtered dimension table is broadcasted over the network).

Scenario 4: Repartition by Column vs. Repartition by Integer Limits

The Scenario

A developer wants to partition a table on disk. They call df.repartition(10, "country"). The output folder has exactly 10 files, but some files are 5GB while others are 2KB.

The Questions

1. Why does repartition(10, "country") generate size-skewed files?
2. How does repartition("country") differ when no partition number is specified?

Detailed Solution & Architectural Analysis

1. Modulo Hash Collisions

When you specify a target number N (10):

• Spark hashes the column value and applies a modulo operation: Hash(country) % 10.
• If your dataset is skewed (e.g. 90% of rows represent "US"), all US records map to the same bucket.
• Furthermore, different countries (e.g., "Germany" and "France") can hash to the same bucket number, creating massive files while other buckets remain empty.

2. Cardinality-Based Partitioning

If you call df.repartition("country") without a count:

• Spark uses the default shuffle partition count (e.g., 200).
• Records are hashed uniformly. This distributes rows evenly but generates up to 200 files per country directory, which can lead to file system bloat. To avoid this, write bucketed tables or use .coalesce() carefully.